/*
Copyright 2010 Dmitry Barashev

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
 */
package net.barashev.seclub.io;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import net.barashev.seclub.io.file.FileManager;
import net.barashev.seclub.io.file.VFile;

import org.apache.hadoop.util.IndexedSortable;
import org.apache.hadoop.util.QuickSort;

/**
 * A few algorithms for working with data records.
 * @author Dmitry Barashev
 */
public class Sorter {
    private static final QuickSort sorter = new QuickSort();

    /**
     * Sorts the content of the given iterating {@code slot} using the given {@comparator}.
     * Writes the result to the same slot.
     */
    public static <T,M extends DataMarshaller<T>> void sort(
            final IteratingSlot<T,M> slot, final Comparator<T> comparator) {
        IndexedSortable sortable = new IndexedSortable() {
            public int compare(int i, int j) {
                T t1 = slot.getBlock(i / slot.getBlockCount()).get(i % slot.getBlockCount());
                T t2 = slot.getBlock(j / slot.getBlockCount()).get(j % slot.getBlockCount());
                return comparator.compare(t1, t2);
            }

            public void swap(int i, int j) {
                FixedLengthRecordBlock<T, DataMarshaller<T>> b1 = slot.getBlock(i / slot.getBlockCount());
                FixedLengthRecordBlock<T, DataMarshaller<T>> b2 = slot.getBlock(j / slot.getBlockCount());

                T t1 = b1.get(i % slot.getBlockCount());
                T t2 = b2.get(j % slot.getBlockCount());

                b2.set(j % slot.getBlockCount(), t1);
                b1.set(i % slot.getBlockCount(), t2);
            }
        };
        sorter.sort(sortable, 0, slot.getRecordCount());
    }

    /**
     * Implements a multiway merge-sort of the given {@code file}.
     * Precondition: there are at least {@code slotSize + 1} available buffers in
     * the {@code bufferMgr}.
     *
     * Creates an iterating slot of {@code slotSize} buffers for sorting chunks of
     * the input file on the first phase. Writes sorted sublists to the temporary files,
     * merges them on the second phase and writes the result to {@code outputFile}.
     *
     * @param file
     * @param outputFile
     * @param comparator
     * @param bufferMgr
     * @param fileMgr
     * @param slotSize
     * @throws IOException
     */
    public static <T> void multiwayMergeSort(
            VFile<T> file, VFile<T> outputFile, Comparator<T> comparator,
            BufferManager bufferMgr, FileManager fileMgr, int slotSize) throws IOException {
        List<VFile<T>> tempFiles = makeSortedSublists(
                file, comparator, bufferMgr, fileMgr, slotSize);
        HashMap<IteratingSlot<T, DataMarshaller<T>>, Integer> mergeSlots = initialPullHeads(
                bufferMgr, tempFiles);

        OutputSlot<T,DataMarshaller<T>> outputSlot = bufferMgr.allocateOutputSlot(outputFile);
        mergeSortedSublists(comparator, mergeSlots, outputSlot);
        outputSlot.flush();
        outputSlot.release();
    }

    private static <T> void mergeSortedSublists(
            Comparator<T> comparator, HashMap<IteratingSlot<T, DataMarshaller<T>>, Integer> mergeSlots,
            OutputSlot<T, DataMarshaller<T>> outputSlot) throws IOException {
        while(!mergeSlots.isEmpty()) {
            T min = findMinAndPull(comparator, mergeSlots);
            if (min != null) {
                outputSlot.append(min);
            }
        }
    }

    public static <T> T findMinAndPull(Comparator<T> comparator,
            HashMap<IteratingSlot<T, DataMarshaller<T>>, Integer> mergeSlots) throws IOException {
        T minRecord = null;
        Map.Entry<IteratingSlot<T,DataMarshaller<T>>, Integer> minEntry = null;
        for (Iterator<Map.Entry<IteratingSlot<T,DataMarshaller<T>>, Integer>> it =
            mergeSlots.entrySet().iterator(); it.hasNext();) {
            Map.Entry<IteratingSlot<T,DataMarshaller<T>>, Integer> entry = it.next();
            IteratingSlot<T, DataMarshaller<T>> slot = entry.getKey();
            Integer idxRecord = entry.getValue();
            if (idxRecord == slot.getRecordCount()) {
                if (slot.pull()) {
                    idxRecord = 0;
                    entry.setValue(0);
                }
                else {
                    it.remove();
                    continue;
                }
            }
            T topRecord = slot.getBlock(0).get(idxRecord);
            if ((minRecord == null) || (comparator.compare(minRecord, topRecord) > 0)) {
                minRecord = topRecord;
                minEntry = entry;
            }
        }
        if (minRecord != null) {
            minEntry.setValue(minEntry.getValue() + 1);
        }
        return minRecord;
    }


    public static <T> HashMap<IteratingSlot<T, DataMarshaller<T>>, Integer> initialPullHeads(
            BufferManager bufferMgr, List<VFile<T>> tempFiles)
            throws IOException {
        HashMap<IteratingSlot<T, DataMarshaller<T>>, Integer> mergeSlots =
            new HashMap<IteratingSlot<T,DataMarshaller<T>>, Integer>();
        for (int i=0; i<tempFiles.size(); i++) {
            IteratingSlot<T, DataMarshaller<T>> mergeSlot = bufferMgr.allocateIteratingSlot(1, tempFiles.get(i));
            boolean pulled = mergeSlot.pull();
            assert pulled;
            mergeSlots.put(mergeSlot, 0);
        }
        return mergeSlots;
    }

    private static <T> List<VFile<T>> makeSortedSublists(VFile<T> file,
            Comparator<T> comparator, BufferManager bufferMgr,
            FileManager fileMgr, int slotSize) throws IOException {
        IteratingSlot<T, DataMarshaller<T>> sortSlot = bufferMgr.allocateIteratingSlot(slotSize, file);
        List<VFile<T>> tempFiles = new ArrayList<VFile<T>>();
        while (sortSlot.pull()) {
            sort(sortSlot, comparator);
            VFile<T> tempFile = fileMgr.createFile(file.getName() + tempFiles.size(), file.getMarshaller());
            OutputSlot<T,DataMarshaller<T>> outputSlot = bufferMgr.allocateOutputSlot(tempFile);
            for (int i=0; i<sortSlot.getBlockCount(); i++) {
                FixedLengthRecordBlock<T,DataMarshaller<T>> block = sortSlot.getBlock(i);
                for (int j=0; j<block.getRecordCount(); j++) {
                    outputSlot.append(block.get(j));
                }
            }
            outputSlot.flush();
            outputSlot.release();
            tempFiles.add(tempFile);
        }
        return tempFiles;
    }

    public static <T> void multiwayMergeSort(
            List<VFile<T>> tempFiles, VFile<T> outputFile, Comparator<T> comparator,
            BufferManager bufferMgr, FileManager fileMgr) throws IOException {
        HashMap<IteratingSlot<T, DataMarshaller<T>>, Integer> mergeSlots = initialPullHeads(
                bufferMgr, tempFiles);

        OutputSlot<T,DataMarshaller<T>> outputSlot = bufferMgr.allocateOutputSlot(outputFile);
        mergeSortedSublists(comparator, mergeSlots, outputSlot);
        outputSlot.flush();
        outputSlot.release();
    }
}
